/**
 * yeganaaa@163.com
 * author: (مۇختەرجان مەخمۇت) muhtaer mahmood
 * 6/1/2020, 3:51 PM GMT+06:00
 */
import {PromiseCompleter, RpcBridgeHost} from "rpc_dream";
import uuid from "uuid"
const Amqp = require("amqplib")
import * as Net from "net";
import {IRpcSocketServer} from "rpc_dream";
import {RpcTcpSocketServer} from "rpc_dream_server";
export enum AmqpSchema {
    AMQP="amqp",
    AMQPS="amqps",
}
export interface IAmqpConfig {
    protocol?: AmqpSchema;
    userName?: string
    password?: string
    host: string
    port: number
    socketOptions?: Net.TcpSocketConnectOpts
}
export interface IConnectionConfig {
    virtualEndPoint: string
}
enum BridgeWorkMode {
    Master,
    Slave
}
export interface ConsumerData {
    fields: {
        "consumerTag": string,
        "deliveryTag": number,
        "redelivered": boolean,
        "exchange": string,
        "routingKey": string
    },
    properties: {
        headers: any,
        deliveryMode: number
    },
    content: Buffer
}
export enum SignalType {
    HELLO_FROM_CLIENT = "Hello?",
    HI_FROM_SERVER = "Hi!",
    CONNECT_REQUEST_FROM_CLIENT = "I want to connect you, I'm a client",
    CONNECTED_RESPONSE_FROM_SERVER = "Okay, you are connected!",
    DISCONNECT_REQUEST = "I want to disconnect",
    DISCONNECTED = "Okay then."
}
export interface Signal {
    signal: SignalType,
    extra: any
    clientExchange: string
}

class EmulatedSocket {
    onMessage: (data: Buffer) => void = null!
    onError: (error: Error) => void = null!
    onConnected: () => void = null!
    onDisconnected: () => void = null!
    open: () => Promise<void> = null!
    close: () => Promise<void> = null!
    clientExchange: string = null!
    send: (data: Buffer) => Promise<void> = null!
}
async function generateEmulatedSocket(amqpConfig: IAmqpConfig, communicationConfig: IConnectionConfig, workMode: BridgeWorkMode, socketReceiver: (socket: EmulatedSocket) => void) {
    const sockets = new Map<string, EmulatedSocket>()
    function buildServerConfig() {
        const config = {
            exchange: `SERVER::${communicationConfig.virtualEndPoint}`,
            queue1: "",
            queue2: "",
        }
        config.queue1 = `${config.exchange}::SIGNAL`
        config.queue2 = `${config.exchange}::DATA`
        return config
    }
    function buildClientConfig(_uuid = uuid()) {
        const config = {
            exchange: `CLIENT::${communicationConfig.virtualEndPoint}::${_uuid.toUpperCase()}`,
            queue1: "",
            queue2: ""
        }
        config.queue1 = `${config.exchange}::SIGNAL`
        config.queue2 = `${config.exchange}::DATA`
        return config
    }
    const queueConfig = {
        durable: false,
        autoDelete: true,
        exclusive: true,
    }
    const exchangeConfig = {
        durable: false,
        autoDelete: true,
    }
    const connection = await Amqp.connect({
        hostname: amqpConfig.host,
        port: amqpConfig.port,
        schema: amqpConfig.protocol ?? AmqpSchema.AMQP,
        username: amqpConfig.userName,
        password: amqpConfig.password,
    }, {}, amqpConfig.socketOptions)
    const mqChannel = await connection.createChannel()
    const preConfig = workMode == BridgeWorkMode.Master ? buildServerConfig() : buildClientConfig()
    const serverConfig = buildServerConfig()
    await mqChannel.assertExchange(preConfig.exchange, "direct", exchangeConfig)
    await mqChannel.assertQueue(preConfig.queue1, queueConfig)
    await mqChannel.assertQueue(preConfig.queue2, queueConfig)
    await mqChannel.bindQueue(preConfig.queue1, preConfig.exchange, "SIGNAL")
    await mqChannel.bindQueue(preConfig.queue2, preConfig.exchange, "DATA")
    const consumer1Tag = await mqChannel.consume(preConfig.queue1, async (signal: ConsumerData) => {
        let content: Signal = null!
        const signalObject = (JSON.parse(signal.content.toString())) as Signal
        if (workMode == BridgeWorkMode.Master) {
            let valid = signalObject.signal == SignalType.HELLO_FROM_CLIENT
            if (valid) {
                content = {
                    signal: SignalType.HI_FROM_SERVER,
                    clientExchange: signalObject.extra,
                    extra: {}
                }
                await publishMessage({
                    options: {persistent: false},
                    content,
                    exchange: signalObject.clientExchange,
                    routeKey: "SIGNAL"
                })
                return
            }
            valid = signalObject.signal == SignalType.CONNECT_REQUEST_FROM_CLIENT
            if (valid) {
                content = {
                    signal: SignalType.CONNECTED_RESPONSE_FROM_SERVER,
                    extra: {},
                    clientExchange: signalObject.clientExchange,
                }
                await publishMessage({
                    content,
                    exchange: content.clientExchange,
                    routeKey: "SIGNAL"
                })
                const socket = new EmulatedSocket()
                socket.clientExchange = signalObject.clientExchange
                socket.send = async (data: Buffer) => {
                    await publishMessage({
                        exchange: socket.clientExchange,
                        content: data,
                        options: {persistent: false, headers: {client: signalObject.clientExchange}},
                        routeKey: "DATA"
                    })
                }
                socket.close = async () => {
                    const content: Signal = {
                        signal: SignalType.DISCONNECT_REQUEST,
                        clientExchange: socket.clientExchange,
                        extra: {}
                    }
                    await publishMessage({
                        routeKey: "SIGNAL",
                        options: {persistent: false},
                        content: content,
                        exchange: socket.clientExchange
                    })
                }
                socket.open = async () => {
                }
                sockets.set(signalObject.clientExchange, socket)
                socketReceiver(socket)
                socket.onConnected()
                return
            }
            const closeSocket = () => {
                const socket = sockets.get(signalObject.clientExchange)
                if (socket != null) {
                    socket.onDisconnected()
                    sockets.delete(signalObject.clientExchange)
                }
            }
            valid = signalObject.signal == SignalType.DISCONNECT_REQUEST
            if (valid) {
                content = {
                    signal: SignalType.DISCONNECTED,
                    extra: {},
                    clientExchange: signalObject.clientExchange
                }
                await publishMessage({
                    content,
                    exchange: signalObject.clientExchange,
                    options: {},
                    routeKey: "SIGNAL"
                })
                closeSocket()
                return
            }
            valid = signalObject.signal == SignalType.DISCONNECTED
            if (valid) {
                closeSocket()
                return
            }
        }
        if (workMode == BridgeWorkMode.Slave) {
            let valid = true
            valid = valid && signalObject.signal == SignalType.HI_FROM_SERVER
            if (valid) {
                content = {
                    signal: SignalType.CONNECT_REQUEST_FROM_CLIENT,
                    clientExchange: preConfig.exchange,
                    extra: {}
                }
                await publishMessage({
                    exchange: serverConfig.exchange,
                    routeKey: "SIGNAL",
                    content
                })
                return
            }
            valid = signalObject.signal == SignalType.CONNECTED_RESPONSE_FROM_SERVER
            if (valid) {
                const socket = new EmulatedSocket()
                socket.clientExchange = preConfig.exchange
                socket.open = async () => {}
                socket.close = async () => {
                    const content: Signal = {
                        signal: SignalType.DISCONNECT_REQUEST,
                        clientExchange: signalObject.clientExchange,
                        extra: {}
                    }
                    await publishMessage({
                        routeKey: "SIGNAL",
                        options: {persistent: false},
                        content: content,
                        exchange: serverConfig.exchange
                    })
                }
                socket.send = async (data: Buffer) => {
                    await publishMessage({
                        exchange: serverConfig.exchange,
                        content: data,
                        options: {persistent: false, headers: {client: preConfig.exchange}},
                        routeKey: "DATA"
                    })
                }
                sockets.set(socket.clientExchange, socket)
                socketReceiver(socket)
                socket.onConnected()
                return
            }
            const closeSocket = () => {
                const socket = sockets.get(preConfig.exchange)
                if (socket != null) {
                    socket.onDisconnected()
                    sockets.delete(signalObject.clientExchange)
                }
            }
            valid = signalObject.signal == SignalType.DISCONNECT_REQUEST
            if (valid) {
                content = {
                    signal: SignalType.DISCONNECTED,
                    extra: {},
                    clientExchange: preConfig.exchange
                }
                await publishMessage({
                    content,
                    exchange: preConfig.exchange,
                    options: {},
                    routeKey: "SIGNAL"
                })
                closeSocket()
                return
            }
            valid = signalObject.signal == SignalType.DISCONNECTED
            if (valid) {
                closeSocket()
                return
            }
        }
    }, {noAck: true})
    const consumer2Tag = await mqChannel.consume(preConfig.queue2, (data: ConsumerData) => {
        const exchange = data.properties.headers.client as string
        const clientSocket = sockets.get(exchange)
        if (clientSocket == null) return
        clientSocket.onMessage(data.content)
    }, {noAck: true})
    if (workMode == BridgeWorkMode.Slave) {
        await publishMessage({
            routeKey: "SIGNAL",
            exchange: serverConfig.exchange,
            content: {
                signal: SignalType.HELLO_FROM_CLIENT,
                clientExchange: preConfig.exchange
            },
            options: {
                persistent: false
            }
        })
    }
    function publishMessage(args: {content: any, exchange: string, routeKey: string, options?: {persistent?: boolean, expiration?: number, headers?: any}}): Promise<any> {
        if (args.content.constructor.name != "Buffer") args.content = Buffer.from(JSON.stringify(args.content))
        return mqChannel.publish(args.exchange, args.routeKey, args.content, args.options);
    }
}
enum InternalConnectionState {
    Disconnected,
    Connected
}
export class AmqpConnectionBridgeHost extends RpcBridgeHost<EmulatedSocket>{
    socket: EmulatedSocket = null!
    private connectionState: InternalConnectionState = InternalConnectionState.Disconnected
    constructor(private amqpConfig: IAmqpConfig, private communicationConfig: IConnectionConfig) {
        super();
    }

    close(callback: () => void): void {
        this.socket.onDisconnected = callback
        this.socket.close()
    }

    async initialize(): Promise<void> {
        const initializer = () => {
            const completer = new PromiseCompleter<void>()
            generateEmulatedSocket(this.amqpConfig, this.communicationConfig, BridgeWorkMode.Slave, socket => {
                this.socket = socket
                this.socket.onConnected = () => {
                    completer.resolve()
                }
            })
            return completer.build()
        }
        await initializer()
        this.socket.onMessage = data => {this.onMessageReceived(data)}
        this.socket.onDisconnected = () => {
            if (this.connectionState == InternalConnectionState.Disconnected) return
            this.connectionState = InternalConnectionState.Disconnected
            this.onClosed()
        }
        this.socket.onError = error => {this.onError(error)}
        this.connectionState = InternalConnectionState.Connected
        this.onConnected()
    }
    send(data: Uint8Array, callback: (error: Error) => void): void {
        this.socket.send(Buffer.from(data)).then(value => {callback(null!)}).catch(reason => {
            callback(reason)
            this.socket.onDisconnected()
        })
    }
}
class PrivateBridgeUsedByTheServer extends RpcBridgeHost<EmulatedSocket>{
    private canBeResume = new PromiseCompleter<void>()
    private connectionState = InternalConnectionState.Disconnected
    constructor(public socket: EmulatedSocket) {
        super();
        this.canBeResume.build()
        this.socket.onConnected = async () => {
            await this.canBeResume.completer
            this.connectionState = InternalConnectionState.Connected
            this.onConnected()
        }
        this.socket.onMessage = data => {this.onMessageReceived(data)}
        this.socket.onDisconnected = () => {
            if (this.connectionState == InternalConnectionState.Disconnected) return
            this.connectionState = InternalConnectionState.Disconnected
            this.onClosed()
        }
        this.socket.onError = error => {this.onError(error)}
    }
    close(callback: () => void): void {
        this.socket.close().then(callback).catch(callback)
    }

    async initialize(): Promise<void> {
        this.canBeResume.resolve();
    }

    send(data: Uint8Array, callback: (error: Error) => void): void {
        this.socket.send(Buffer.from(data)).then(value => {callback(null!)}).catch(reason => {
            this.socket.onDisconnected()
            callback(reason)
        })
    }
}
export class AmqpConnectionBridgeServer implements IRpcSocketServer<any> {
    server: any;
    onClientConnected: (client: RpcBridgeHost<any>) => void = null!
    onLaunch: () => void = null!
    onShutDown: () => void = null!

    constructor(private amqpConfig: IAmqpConfig, private communicationConfig: IConnectionConfig) {
    }

    onError(error: Error): void {
        console.error(error)
    }

    public async prepareForLaunch(): Promise<void> {
        await generateEmulatedSocket(this.amqpConfig, this.communicationConfig, BridgeWorkMode.Master, socket => {
            const bridge = new PrivateBridgeUsedByTheServer(socket)
            this.onClientConnected(bridge)
        })
        this.onLaunch()
    }
    async shutdown(): Promise<void> {
        throw new Error("Unimplemented exception")
    }
}